#ifndef KPEXTSYNCMISC_H
#define KPEXTSYNCMISC_H

#include <iostream>
#include <stdlib.h>
#include <memory>
#include <queue>
#include <atomic>
#include <condition_variable>
#include <unordered_set>
#include <unordered_map>
#include <chrono>
#include <thread>
#include <future>

/**
 * @brief 使用队列实现的模板化信号量类
 */
template<typename T>
class SemaphoreQueT{
public:
    explicit SemaphoreQueT();


    virtual ~SemaphoreQueT();

    /**
     * @brief 请求内部队列保持的一个资源（挂起线程，直到资源可用或被强制唤醒）
     * @param defValue  如果挂起线程被非获取资源条件唤醒, 返回的默认值
     * @return 返回资源
     */
    virtual T acquire(T defValue);

    /**
     * @brief 获取内部队列可用的资源数量
     * @return 内部队列可用的资源数量
     */
    virtual int available();

    /**
     * @brief 释放一个资源到内部队列
     * @param res  要加入到内部队列的资源
     */
    virtual void release(T & res);

    /**
     * @brief 尝试请求内部队列保持的一个资源
     * @param result
     * @return
     */
    virtual bool tryAcquire(T * result);

    /**
     * @brief tryAcquire
     * @param timeout_ms
     * @param result
     * @return
     */
    virtual bool tryAcquire(int timeout_ms, T * result);

    /**
     * @brief 一般用于wakeupAll函数用前，设置唤醒后退出标记
     * @param b  唤醒后退出标记新的值
     */
    virtual void setIsWakeupToQuit(bool b = false);


    /**
     * @brief 强制唤醒所有等待资源而挂起的线程
     */
    virtual void wakeupAll();

    /**
     * @brief 一般用于wakeupAll函数用后，获取唤醒后退出标记
     * @return 如果是唤醒后退出返回true
     */
    virtual bool getIsWakeupToQuit();

    /**
     * @brief 清除所有在内部队列中的资源
     * @param destroyHandle 销毁的回调函数
     */
    virtual void clearAllResource(void (* destroyHandle)(T) = nullptr);

protected:
    /**
     * @brief 用于同步的条件变量
     */
    std::condition_variable m_cv;

    /**
     * @brief 队列资源访问锁
     */
    std::mutex m_queMutex;

    /**
     * @brief 队列资源
     */
    std::queue<T> m_queue;

    /**
     * @brief 唤醒后退出标记
     */
    std::atomic<bool> m_IsWakeupToQuit;

}; // semaphoreQueT

#endif // KPEXTSYNCMISC_H

//---------------------------------------semaphoreQueT -----------------------------------------------

template<typename T>
SemaphoreQueT<T>::SemaphoreQueT():m_queue()
{
    m_IsWakeupToQuit.store(false);
}

template<typename T>
SemaphoreQueT<T>::~SemaphoreQueT()
{
//    {
//        std::lock_guard<std::mutex> lk(m_queMutex);
//    }
}

template<typename T>
T SemaphoreQueT<T>::acquire(T defValue)
{
    // unique_lock与lock_guard功能相似， 但提供手动锁定、释放的自由
    std::unique_lock<std::mutex> lk(m_queMutex);
    const auto & cond = [&]() -> bool {
        return m_queue.size() > 0 || m_IsWakeupToQuit.load();
    };

    m_cv.wait(lk, cond);
    // wait() 执行过程中，锁定锁，判定资源是否满足，不满足就解锁挂起线程
    // wait() 返回后， m_queMutex锁是锁定的且条件已经满足
    if(!m_IsWakeupToQuit.load() && m_queue.size() > 0){
        auto ret = m_queue.front();
        m_queue.pop();
        return ret;
    }else{
        return defValue;
    }
}

template<typename T>
int SemaphoreQueT<T>::available()
{
    int ret = 0;
    {
        std::lock_guard<std::mutex> lk(m_queMutex);
        ret = static_cast<int>(m_queue.size());
    }
    return ret;
}

template<typename T>
void SemaphoreQueT<T>::release(T & res)
{
    {
        std::lock_guard<std::mutex> lk(m_queMutex);
        m_queue.push(res);
    }
    //notify_one前应解锁资源锁
    m_cv.notify_one();
}

template<typename T>
bool SemaphoreQueT<T>::tryAcquire(T * result)
{
    bool ret = false;
    if(m_queMutex.try_lock() && result){
        if(m_queue.size() > 0){
            *result = m_queue.front();
            m_queue.pop();
            ret = true;
        }
        m_queMutex.unlock();
    }

    return ret;
}

template<typename T>
bool SemaphoreQueT<T>::tryAcquire(int timeout_ms, T * result)
{
    using  chronoms = std::chrono::milliseconds;
    bool ret = false;
    std::unique_lock<std::mutex> lk(m_queMutex);
    auto time = chronoms(timeout_ms);
    const auto & cond = [&]() -> bool{
        return m_queue.size() > 0 || m_IsWakeupToQuit.load();
    };

    if(m_cv.wait_for(lk, time, cond)){
        if(!m_IsWakeupToQuit.load() && m_queue.size() > 0){
            *result = m_queue.front();
            m_queue.pop();
            ret = true;
        }
    }
    return ret;
}

template<typename T>
void SemaphoreQueT<T>::wakeupAll()
{
    m_cv.notify_all();
}

template<typename T>
bool SemaphoreQueT<T>::getIsWakeupToQuit()
{
    return m_IsWakeupToQuit.load();
}

template<typename T>
void SemaphoreQueT<T>::clearAllResource(void (* destroyHandle)(T))
{
    std::lock_guard<std::mutex> lk(m_queMutex);
    while(!m_queue.empty()){
        auto v = m_queue.front();
        m_queue.pop();
        if(destroyHandle){
            destroyHandle(v);
        }
    }
}

template<typename T>
void SemaphoreQueT<T>::setIsWakeupToQuit(bool b)
{
    m_IsWakeupToQuit.store(b);
}

//--------------------------------------------------------------------------------------------------

//---------------------------------------semWrapNtoOneT -----------------------------------------------

/**
 * @brief 使用双队列实现的 多输入折叠为单输出的信号量类(FIFO)
 */
template<typename T_in, typename T_out>
class SemWrapNtoOneT{
public:
    /**
     * @brief 构造函数
     * @param wrapN  输入资源折叠， 就是保证多个资源放入同一个线程
    */
    explicit SemWrapNtoOneT(uint32_t wrapN) :
        m_wrapN(wrapN),
        m_wrapIdx(0),
        m_allThreadsRegistered(false),
        m_threadIdx(0),
        m_outIdx(0)
    {
        if(m_wrapN == 0) m_wrapN = 1;
        m_IsWakeupToQuit.store(false);
    }

    virtual ~SemWrapNtoOneT(){}


    /**
     * @brief 录入一个要进行同步的线程ID，应在所有要同步的线程运行前注册它们
     * @param id 线程ID
     */
    virtual void registerThreadID(const std::thread::id & id){
        std::lock_guard<std::mutex> lk(m_inQueMutex);
        auto it = std::find(m_syncThreads.begin(),
                            m_syncThreads.end(),
                            id
                            );
        if(it == m_syncThreads.end()){
            m_syncThreads.push_back(id);
            m_inCondVars.push_back(std::make_shared<std::condition_variable>());
            m_inQueues.push_back(std::queue<T_in>());
            m_outQueues.push_back(std::queue<T_out>());
        }
    }

    /**
     * @brief 等待所有需要进行同步的线程注册完成（可设置超时）
     * @param timeout_ms  超时时长，单位为毫秒; 小于0零表示等待0x7fffFFFF毫秒
     */
    virtual bool waitAllThreadsRegistered(int timeout_ms = 0x7FFFFFFF){
        using  chronoms = std::chrono::milliseconds;
        bool ret = false;
        if(timeout_ms < 0) timeout_ms = 0x7FFFFFFF;
        auto time = chronoms(timeout_ms);

        const auto & condfunc = [this]()
                -> bool{
            return m_allThreadsRegistered;
        };

        std::unique_lock<std::mutex> lk(m_outQueMutex);
        if(m_regThreadCondVar.wait_for(lk, time, condfunc)){
            ret = m_allThreadsRegistered;
        }
        return ret;
    }

    /**
     * @brief 标记所有需要注册的线程已经注册完成
     */
    virtual void setAllThreadsRegistered(){
        {
            std::lock_guard<std::mutex> lk(m_outQueMutex);
            m_allThreadsRegistered = true;
        }
        m_regThreadCondVar.notify_all();
    }

    /**
     * @brief 向输入队列压入一项数据
     * @param in 压入的数据
     */
    virtual void pushInput(T_in & in){
        std::shared_ptr<std::condition_variable> cond(nullptr);
        {
            std::lock_guard<std::mutex> lk(m_inQueMutex);

            m_inQueues[m_threadIdx].push(in);
            cond = m_inCondVars[m_threadIdx];

            m_wrapIdx++;
            if(m_wrapIdx >= m_wrapN){
                m_wrapIdx = 0;
                m_threadIdx++;
            }

            uint32_t threadcnt = static_cast<uint32_t>(m_syncThreads.size());
            if(m_threadIdx >= threadcnt){
                m_threadIdx = 0;
            }
        }

        if(cond){
            cond->notify_one();
        }
    }

    /**
     * @brief 尝试从输入队列获取一项数据（可设置超时）
     * @param res 返回的数据， 由调用方申请和释放内存
     * @param timeout_ms 超时时长，单位为毫秒; 小于0零表示等待0x7fffFFFF毫秒
     * @return  成功获取返回true， 否则返回false
     */
    virtual bool acquireInput(T_in * res, int timeout_ms = 0x7FFFFFFF){
        using  chronoms = std::chrono::milliseconds;
        bool ret = false;
        uint32_t threadindex = 0;
        std::shared_ptr<std::condition_variable> cv(nullptr);

        if(!res){
            return ret;
        }


        {
            auto threadid = std::this_thread::get_id();
            uint32_t threadcnt = static_cast<uint32_t>(m_syncThreads.size());
            for (uint32_t i = 0; i < threadcnt; ++i) {
                if(m_syncThreads[i] == threadid){
                    threadindex = i;
                    cv = m_inCondVars[threadindex];
                    break;
                }
            }
        }

        if(!cv){
            return false;
        }

        if(timeout_ms < 0) timeout_ms = 0x7FFFFFFF;
        auto time = chronoms(timeout_ms);

        const auto & condfunc = [&, threadindex]()
                -> bool{
            auto & que = m_inQueues[threadindex];
            return que.size() > 0 || m_IsWakeupToQuit.load();
        };

        std::unique_lock<std::mutex> lk(m_inQueMutex);
        if(cv->wait_for(lk, time, condfunc)){
            auto & que = m_inQueues[threadindex];
            if(!m_IsWakeupToQuit.load() && que.size() > 0){
                *res = que.front();
                que.pop();
                ret = true;
            }
        }

        return ret;
    }

    /**
     * @brief 向输出队列压入一项数据
     * @param out 压入的数据
     */
    virtual void pushOutput(T_out & out){
        uint32_t threadindex = ~0U;
        {
            auto threadid = std::this_thread::get_id();
            uint32_t threadcnt = static_cast<uint32_t>(m_syncThreads.size());
            for (uint32_t i = 0; i < threadcnt; ++i) {
                if(m_syncThreads[i] == threadid){
                    threadindex = i;
                    break;
                }
            }
        }

        if(threadindex == ~0U) return;

        {
            std::lock_guard<std::mutex> lk(m_outQueMutex);
            m_outQueues[threadindex].push(out);
        }
        m_outCondVar.notify_one();
    }

    /**
     * @brief 尝试从输出队列获取一项数据（可设置超时）
     * @param res 返回的数据， 由调用方申请和释放内存
     * @param timeout_ms 超时时长，单位为毫秒; 小于0零表示等待0x7fffFFFF毫秒
     * @return  成功获取返回true， 否则返回false
     */
    virtual bool acquireOutput(T_out * res, int timeout_ms = 1){
        using  chronoms = std::chrono::milliseconds;
        bool ret = false;
        if(timeout_ms < 0) timeout_ms = 0x7FFFFFFF;
        auto time = chronoms(timeout_ms);

        const auto & condfunc = [&]()
                -> bool{
            auto & que = m_outQueues[m_outIdx];
            return que.size() > 0 || m_IsWakeupToQuit.load();
        };

        std::unique_lock<std::mutex> lk(m_outQueMutex);
        if(m_outCondVar.wait_for(lk, time, condfunc)){
            auto & que = m_outQueues[m_outIdx];
            if(que.size() > 0 && !m_IsWakeupToQuit.load()){
                *res = que.front();
                que.pop();
                m_outIdx++;
                uint32_t threadcnt = static_cast<uint32_t>(m_syncThreads.size());
                if(m_outIdx >= threadcnt){
                    m_outIdx = 0;
                }
                ret = true;
            }
        }

        return ret;
    }

    /**
     * @brief 判断是否有输出数据项
     * @return 有返回true, 无返回false
     */
    virtual bool hasOutput(){
        bool ret = false;
        {
            std::lock_guard<std::mutex> lk(m_outQueMutex);
            auto & que = m_outQueues[m_outIdx];
            ret = que.size() > 0;
        }

        return ret;
    }

    /**
     * @brief 一般用于wakeupAll函数用前，设置唤醒后退出标记
     * @param b  唤醒后退出标记新的值
     */
    virtual void setIsWakeupToQuit(bool b)
    {
        std::lock_guard<std::mutex> lki(m_inQueMutex);
        std::lock_guard<std::mutex> lko(m_outQueMutex);

        m_IsWakeupToQuit.store(b);
    }

    /**
     * @brief 强制唤醒所有等待资源而挂起的要被同步的线程
     */
    virtual void wakeupAll(){
        std::lock_guard<std::mutex> lki(m_inQueMutex);
        std::lock_guard<std::mutex> lko(m_outQueMutex);

        for(auto & cv : m_inCondVars){
            cv->notify_all();
        }
        m_outCondVar.notify_all();
    }

    /**
     * @brief 一般用于wakeupAll函数用后，获取唤醒后退出标记
     * @return 如果是唤醒后退出返回true
     */
    virtual bool getIsWakeupToQuit()
    {
        return m_IsWakeupToQuit.load();
    }

protected:

    /**
     * @brief 折叠次数
     */
    uint32_t m_wrapN;

    /**
     * @brief 当前的折叠索引
     */
    uint32_t m_wrapIdx;

    /**
     * @brief 用于等待所有线程注册完成的条件变量
     */
    std::condition_variable m_regThreadCondVar;

    /**
     * @brief 所有线程注册完成标记
     */
    bool m_allThreadsRegistered;

    /**
     * @brief 输入队列资源访问锁
     */
    std::mutex m_inQueMutex;

    /**
     * @brief 输入输出队列资源访问锁
     */
    std::mutex m_outQueMutex;

    /**
     * @brief 输入队列资源
     */
    std::vector<std::queue<T_in>> m_inQueues;

    /**
     * @brief 输出队列资源
     */
    std::vector<std::queue<T_out>> m_outQueues;

    /**
     * @brief 要进行同步的
     */
    std::vector<std::thread::id> m_syncThreads;

    /**
     * @brief 要进行同步的线程的条件变量
     */
    std::vector<std::shared_ptr<std::condition_variable>>
    m_inCondVars;

    /**
     * @brief 用于输出的条件变量
     */
    std::condition_variable m_outCondVar;

    /**
     * @brief 切换同步的线程计数索引
     */
    uint32_t m_threadIdx;

    /**
     * @brief 输出队列索引
     */
    uint32_t m_outIdx;

    /**
     * @brief 唤醒后退出标记
     */
    std::atomic<bool> m_IsWakeupToQuit;


}; //class semWrapNtoOneT


//--------------------------------------------------------------------------------------------------

